package com.imooc.flink.mysql;

import com.imooc.flink.util.MySQLUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.types.Row;

import java.sql.Connection;
import java.sql.PreparedStatement;

/**
 * @author: Chen Yixing
 * @since: 2023/10/16 14:55:12
 * @description:
 */
public class MysqlSink extends RichSinkFunction<Row> {
    Connection connection;
    PreparedStatement insertPstmp;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = MySQLUtils.getConnection();
        insertPstmp = connection.prepareStatement("insert into  t_data_center_test(name) values (?)");
    }

    /**
     * 来一条数据就执行一次
     * <p>
     * 1000w的数据  1000w次
     */
    @Override
    public void invoke(Row row, Context context) throws Exception {
        insertPstmp.setString(1, (String) row.getField(1));
        insertPstmp.execute();
    }

    @Override
    public void close() throws Exception {
        if (insertPstmp != null) {
            insertPstmp.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
